package com.mango.ch08;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import com.mango.HDFSTools.HDFSAPI;

import scala.Tuple2;

/**
 * 默写的spark实现共同好友
 * 
 * @param args
 */
public class MyTest {

	static Path inputPath = new Path("/ComFriendsJob/spark_input");// input
	static Path outPath = new Path("/ComFriendsJob/spark_output");

	@SuppressWarnings({ "serial", "unused", "resource" })
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 创建输入文件夹
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			// hdfs.createDirectory(transactions);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			hdfs.closeFilseSystem();
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		SparkConf sconf = new SparkConf();
		sconf.setAppName("spark_comfriendjob");
		JavaSparkContext jsc = new JavaSparkContext(sconf);
		// 读取数据文本
		JavaRDD<String> lines = jsc.textFile(inputPath.toString());
		// 创建业务需要键值对
		JavaPairRDD<Tuple2<Long, Long>, List<Long>> pairs = lines
				.flatMapToPair(new PairFlatMapFunction<String, Tuple2<Long, Long>, List<Long>>() {

					@Override
					public Iterator<Tuple2<Tuple2<Long, Long>, List<Long>>> call(String t) throws Exception {
						// TODO Auto-generated method stub
						List<Tuple2<Tuple2<Long, Long>, List<Long>>> result = new ArrayList<Tuple2<Tuple2<Long, Long>, List<Long>>>();
						String[] tokens = t.split(","); // ',' 前是用户，后是该用户的好友列表
						Long user = Long.valueOf(tokens[0]);
						String[] friends = tokens[1].split(" ");
						if (friends.length == 1) {
							// 该用户只有一个好友，则用户与该好友之间没有其他的共同好友
							Tuple2<Long, Long> key = buildSortedTuple(user, Long.valueOf(friends[0]));
							return Arrays.asList(new Tuple2<Tuple2<Long, Long>, List<Long>>(key, new ArrayList<>()))
									.iterator();
						}
						List<Long> friendList = new ArrayList<>();
						for (String f : friends) {
							friendList.add(Long.valueOf(f));
						}
						for (Long fid : friendList) {
							Tuple2<Long, Long> key = new Tuple2<Long, Long>(user, fid);
							result.add(new Tuple2<Tuple2<Long, Long>, List<Long>>(key, friendList));

						}
						return result.iterator();
					}

				});
		// 构建好友共同好友列表 方案一
		// 先规约values
		JavaPairRDD<Tuple2<Long, Long>, Iterable<List<Long>>> finalPairs = pairs.groupByKey();
		// 然后构建共同好友(输入的数据类型可自定义）
		JavaPairRDD<Tuple2<Long, Long>, List<Long>> finalResult = finalPairs
				.mapValues(new Function<Iterable<List<Long>>, List<Long>>() {

					@Override
					public List<Long> call(Iterable<List<Long>> v1) throws Exception {
						// TODO Auto-generated method stub
						int size = 0;
						Map<Long, Integer> fm = new HashMap<Long, Integer>();
						List<Long> friList = new ArrayList<>();
						for (List<Long> list : v1) {
							size += 1;
							// 好友列表为空，跳过本次循环
							if (list.isEmpty())
								continue;
							for (Long fid : list) {
								if (fm.get(fid) == null)
									fm.put(fid, 1);
								else {
									int count = fm.get(fid);
									fm.put(fid, ++count);
								}
							}
						}
						// 遍历map 分析是否存在共同好友
						// size=某fid的值 怎表示该fid是一个共同的好友
						for (Entry<Long, Integer> entry : fm.entrySet()) {
							if (entry.getValue() == size)
								friList.add(entry.getKey());
						}

						return friList;
					}
				});
		// 查找共同好友 方案二
		JavaPairRDD<Tuple2<Long, Long>, List<Long>> finalResult2 = pairs
				.reduceByKey(new Function2<List<Long>, List<Long>, List<Long>>() {

					@Override
					public List<Long> call(List<Long> v1, List<Long> v2) throws Exception {
						// TODO Auto-generated method stub
						// 求两个值集的交集
						Set<Long> comFri = new HashSet<>(v1);
						List<Long> result = new ArrayList<>();
						for (Long fid : v2) {
							if (comFri.contains(fid))
								result.add(fid);
						}
						return result;
					}
				});

	}

	protected static List<Long> iterableToList(Iterable<Long> iter) {
		// TODO Auto-generated method stub
		List<Long> list = new ArrayList<>();
		Iterator<Long> it = iter.iterator();
		while (it.hasNext()) {
			list.add(it.next());
		}
		return list;
	}

	protected static Tuple2<Long, Long> buildSortedTuple(Long p1, Long p2) {
		// TODO Auto-generated method stub
		if (p1 > p2) {
			return new Tuple2<Long, Long>(p2, p1);
		} else
			return new Tuple2<Long, Long>(p1, p2);
	}

}
